package com.zyy.joseata.mq;

import com.zyy.joseata.constant.TxEventTypeConst;
import com.zyy.joseata.socket.SocketClient;
import com.zyy.joseata.util.ThreadLocalUtil;
import lombok.extern.slf4j.Slf4j;
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.annotation.Around;
import org.aspectj.lang.annotation.Aspect;
import org.springframework.amqp.core.Message;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.nio.charset.StandardCharsets;

/**
 * @author zhangyy
 */
@Aspect
@Component
@Slf4j
public class RabbitListenerAspect {

    @Resource
    private SocketClient socketClient;

    @Around("@annotation(org.springframework.amqp.rabbit.annotation.RabbitListener) && args(message,..)")
    public void aroundRabbitListener(ProceedingJoinPoint joinPoint, Message message) throws Throwable {

        log.info("RabbitListenerAspect => ThreadName:{}, Around设置前获取的txId:{}",Thread.currentThread().getName(),ThreadLocalUtil.getTxId());

        String msg = new String(message.getBody(), StandardCharsets.UTF_8);
        log.info("RabbitListenerAspect => Received message: {}, txId:{}",msg,message.getMessageProperties().getMessageId());
        String txId = message.getMessageProperties().getMessageId();
        ThreadLocalUtil.setTxId(txId);
        try {
            joinPoint.proceed();
            socketClient.send(TxEventTypeConst.MQ_TX_CONSUME_SUCCESS,msg);
        } catch (Throwable e) {
            socketClient.send(TxEventTypeConst.MQ_TX_CONSUME_FAIL,msg);
            log.error("消费失败",e);
            throw e;
        } finally {
            ThreadLocalUtil.removeTxId();
        }
    }
}
